package com.yujiahao;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

/**
 * @Description:Flink作为kinesis的生产者
 * @Author: yujiahao
 * @Date: 2024/7/10 15:29
 * @Version: 1.0
 */

public class FlinkKinesisProduct {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> data = env.addSource(

                new RichParallelSourceFunction<String>() {
                    @Override
                    public void run(SourceContext<String> ctx) throws Exception {
                        while (true) {
                            Thread.sleep(5000);

                            //浏览量  pv api   先测一个用户，接着测两个用户
                            ctx.collect("{\"user_id\":1,\"course_id\":1,\"lesson_id\":1,\"exercise_id\":1,\"question_id\":1,\"action_type\":\"backend_explain\"}\n" +
                                    "{\"user_id\":1,\"course_id\":1,\"lesson_id\":1,\"exercise_id\":2,\"question_id\":1,\"action_type\":\"backend_explain\"}\n" +
                                    "{\"user_id\":1,\"course_id\":1,\"lesson_id\":1,\"exercise_id\":1,\"question_id\":1,\"action_type\":\"backend_explain\"}\n" +
                                    "{\"user_id\":1,\"course_id\":1,\"lesson_id\":1,\"exercise_id\":3,\"question_id\":1,\"action_type\":\"backend_explain\"}\n" +
                                    "{\"user_id\":1,\"course_id\":1,\"lesson_id\":1,\"exercise_id\":2,\"question_id\":1,\"action_type\":\"backend_explain\"}");

                        }
                    }

                    @Override
                    public void cancel() {

                    }
                }
        );
        data.print("看一下");
        //发送数据到kinesis流中
        data.addSink(KinesisUtil.sendToKinesis("bigdata-dev"));
        env.execute();
    }
}
